package org.example.message_queue.mqserver.core;

import org.example.message_queue.common.Consumer;
import org.example.message_queue.common.ConsumerEnv;
import org.example.message_queue.common.MqException;
import org.example.message_queue.mqserver.VirtualHost;

import java.util.Queue;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

public class ConsumerManager {
//    相当于持有这个元素的父亲元素
    private VirtualHost parent;
//    线程池执行具体的回调任务
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);
//    存放令牌的队列
    private BlockingDeque<String> tokenQueue = new LinkedBlockingDeque<>();
//    扫描线程
    private Thread scannerThread = null;
//      初始化
    public ConsumerManager(VirtualHost p){
        parent = p;
        scannerThread = new Thread(()->{
            while (true){
                try {
//                从令牌队列中获取队列名字
                    String queueName = tokenQueue.take();
//
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
//
                    if (queue == null){
                        throw new MqException("[ConsumerManager] 取令牌后发现，该队列名不存在！queueName=" + queueName);
                    }
//
                    synchronized (queue){
                        consumerMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
//        设置为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    public void notifyConsumer(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
//        找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null){
            throw new MqException("");
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue){
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                consumerMessage(queue);
            }
        }

    }

    private void consumerMessage(MSGQueue queue) {
//        1.按照轮训的方式，找出消费者出来（挑选一个订阅者）
        ConsumerEnv luckyConsumer = queue.chooseConsumer();
        if (luckyConsumer == null){
//            当前队列没有消费者，暂时不消费，等后面有消费者出现再说
            return;
        }
//        2.从队列中获取消息
        Message message = parent.getMemoryDataCenter().getMessage(queue.getName());
        if (message == null){
            return;
        }

//        3.将消息代入消费者的回调方法中丢给线程池执行
        workerPool.submit(() ->{
            try{
//            将消息放入待确认集合中
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(),message);
                luckyConsumer.getConsumer().handleDelivery(luckyConsumer.getConsumerTag(),message.getBasicProperties(),message.getBody());
//            “自动应答”
                if (luckyConsumer.isAutoAck()){
                    if (message.getDeliverMode() == 2){
                        parent.getDiskDataCenter().deleteMessage(queue,message);
                    }
//                删除待确认集合中的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息别成功消费! queueName=" + queue.getName());
                }else {
//            “手动应答”
                    parent.basicAck(queue.getName(),message.getMessageId());
                }
            }catch (Exception e){
                e.printStackTrace();
            }

        });
    }
}
